/**
 * 
 */
package edu.umd.clip.lm.model.data;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.locks.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class EagerTrainingDataNodeReader extends TrainingDataNodeReader {
	private final TrainingDataNodeReader backend;
	private final Queue<ReadableTrainingDataNode> nodes;
	private final Queue<Long> nodeSizes;
	private final long maxTotalSize;
	public static final long DEFAULT_MAX_TOTAL_SIZE = 10 * 1024 * 1024; // 10 MB
	private final Lock lock;
	private final Condition gotSome;
	private final Condition canGetSome;
	private long currentTotalSize;
	private boolean finished;
	private AsyncReaderHelper readerThread;
	
	/**
	 * @param backend
	 * @param maxTotalSize
	 */
	public EagerTrainingDataNodeReader(TrainingDataNodeReader backend, long maxTotalSize) {
		this.backend = backend;
		this.maxTotalSize = maxTotalSize;
		this.nodes = new LinkedList<ReadableTrainingDataNode>();
		this.nodeSizes = new LinkedList<Long>();
		this.lock = new ReentrantLock();
		this.gotSome = lock.newCondition();
		this.canGetSome = lock.newCondition();
		this.currentTotalSize = 0;
		this.finished = false;
		this.readerThread = new AsyncReaderHelper();
		this.readerThread.start();
	}

	
	/**
	 * @param backend
	 */
	public EagerTrainingDataNodeReader(TrainingDataNodeReader backend) {
		this(backend, DEFAULT_MAX_TOTAL_SIZE);
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.model.data.TrainingDataNodeReader#getNext()
	 */
	@Override
	public ReadableTrainingDataNode getNext() {
		try {
			lock.lock();
			while(true) {
				ReadableTrainingDataNode node = nodes.poll();
				if (node != null) {
					long nodeSize = nodeSizes.poll();
					currentTotalSize -= nodeSize;
					if (currentTotalSize < maxTotalSize) {
						canGetSome.signal();
					}
					return node;
				}
				if (finished) return null;
				try {
					gotSome.await();
				} catch (InterruptedException e) {}
			}
		} finally {
			lock.unlock();
		}
	}
	
	private class AsyncReaderHelper extends Thread implements Observer {
		ReadableTrainingDataNode waitingForNode = null;
		
		/**
		 * 
		 */
		public AsyncReaderHelper() {
			super("node read ahead");
			setDaemon(true);
			waitingForNode = null;
			backend.addFinishedReadingNodeObserver(this);
		}

		public void run() {
			try {
				while (true) {
					lock.lock();
					try {
						while (currentTotalSize >= maxTotalSize || waitingForNode != null) {
							try {
								canGetSome.await();
							} catch (InterruptedException e) {
								finished = true;
								gotSome.signal();
								return;
							}
						}
					} finally {
						lock.unlock();
					}
					ReadableTrainingDataNode node = backend.getNext();
					if (node == null) {
						lock.lock();
						finished = true;
						gotSome.signal();
						lock.unlock();
						return;
					}
					long nodeSize = 0;
					for(ReadableTrainingData data : node.getData()) {
						data.hasNext();
						data.parseBuffer();
						if (!data.isFinished()) {
							// if the data does not fit into buffer, give up reading ahead
							nodeSize = maxTotalSize;
							waitingForNode = node;
							//System.err.printf("Waiting for node %d\n", node.getNodeId());
							break;
						}
						nodeSize += data.getBufferSize();
					}
					lock.lock();
					nodes.add(node);
					nodeSizes.add(nodeSize);
					currentTotalSize += nodeSize;
					gotSome.signal();
					lock.unlock();
				}
			} finally {
				backend.deleteFinishedReadingNodeObserver(this);
			}
		}

		/* (non-Javadoc)
		 * @see java.util.Observer#update(java.util.Observable, java.lang.Object)
		 */
		@Override
		public void update(Observable o, Object arg) {
			ReadableTrainingDataNode node = (ReadableTrainingDataNode) arg;
			lock.lock();
			if (waitingForNode == node) {
				waitingForNode = null;
				canGetSome.signal();
			}
			lock.unlock();
			//finishedReadingNode(node);
		}
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.model.data.TrainingDataNodeReader#close()
	 */
	@Override
	public void close() throws IOException {
		lock.lock();
		finished = true;
		AsyncReaderHelper thread = readerThread;
		readerThread = null;
		if (thread != null) {
			thread.interrupt();
		}
		lock.unlock();
		
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		backend.close();
	}
}
